package com.qianfeng.reactor;

//
//                            _ooOoo_  
//                           o8888888o  
//                           88" . "88  
//                           (| -_- |)  
//                            O\ = /O  
//                        ____/`---'\____  
//                      .   ' \\| |// `.  
//                       / \\||| : |||// \  
//                     / _||||| -:- |||||- \  
//                       | | \\\ - /// | |  
//                     | \_| ''\---/'' | |  
//                      \ .-\__ `-` ___/-. /  
//                   ___`. .' /--.--\ `. . __  
//                ."" '< `.___\_<|>_/___.' >'"".  
//               | | : `- \`.;`\ _ /`;.`/ - ` : | |  
//                 \ \ `-. \_ __\ /__ _/ .-` / /  
//         ======`-.____`-.___\_____/___.-`____.-'======  
//                            `=---='  
//  
//         .............................................  
//                  佛祖镇楼                  BUG辟易  
//          佛曰:  
//                  写字楼里写字间，写字间里程序员；  
//                  程序人员写程序，又拿程序换酒钱。  
//                  酒醒只在网上坐，酒醉还来网下眠；  
//                  酒醉酒醒日复日，网上网下年复年。  
//                  但愿老死电脑间，不愿鞠躬老板前；  
//                  奔驰宝马贵者趣，公交自行程序员。  
//                  别人笑我忒疯癫，我笑自己命太贱；  
//  


import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.CountDownLatch;

/**
 * Created by Jackiechan on 2022/7/20 20:30
 *
 * @author Jackiechan
 * @version 1.0
 * @since 1.0
 */
public class TestFlux {


    public static void main(String[] args) throws InterruptedException {
//        Flux.just("1", "2", "3")//根据指定的数据来创建一个flux
//                .subscribe(System.err::println);//订阅数据,收到数据之后我们输出一下,不订阅上面的代码不会执行

//        Flux.create(fluxSink -> {
//            //创建数据
//            fluxSink.next("第一个数据").next("第二个数据").next("第三个数据");
//        }).subscribe(System.err::println);

//        Flux.generate(synchronousSink -> {
//            synchronousSink.next("第一个");//只能调用一次next方法,否则会报错
////            synchronousSink.next("第一个");
////            synchronousSink.next("第一个");
//            synchronousSink.complete();//通知数据完成,消费者就会执行完成方法
//        }).subscribe(System.err::println);


//        Flux.from(Flux.just("a","dasdas","dasdas","dasdasdas")).subscribe(System.err::println);
//        String[] ss = {"zhangsan", "lisi", "wangwu", "zhaoliu"};
//        Flux.fromArray(ss).subscribe(System.err::println);

        List<String> list = new ArrayList<>();
        for (int i = 0; i < 10; i++) {
            list.add("数据q" + i);
        }
//        Flux.fromIterable(list).subscribe(System.err::println);

        // Flux.fromStream(list.stream()).subscribe(System.err::println);
//        Flux.defer(() -> Flux.just("just", "just1", "just2"))
//                .subscribe(System.out::println);
//
//        Flux.just("just", "just1", "just2").subscribe(new Subscriber<String>() {
//            private Subscription subscription;
//            @Override
//            public void onSubscribe(Subscription s) {
//                System.err.println(s);
//                subscription = s;
//                //开始要数据,不要就不会收到
//                subscription.request(1);
//            }
//
//            @Override
//            public void onNext(String s) {
//                //拿到下一个数据的时候
//                System.err.println("dasdasdasdasdas===>" + s);
//                //继续要数据,在数据完成之前,不执行request不会获取到数据,这就是背压
//                subscription.request(1);
//            }
//
//            @Override
//            public void onError(Throwable t) {
//            //发生异常的时候
//                System.err.println("TestFlux中的方法onError 执行了--->");
//            }
//
//            @Override
//            public void onComplete() {
//            //订阅完成的时候
//                System.err.println("TestFlux中的方法onComplete 执行了--->");
//            }
//        });
//
//        Flux.interval(Duration.of(500, ChronoUnit.MILLIS))
//                .subscribe(System.out::println);
//            //因为是异步操作,所以此处需要阻塞



//        Flux.empty().subscribe(System.out::println);
//        Flux.error(new RuntimeException()).subscribe(System.out::println);
//        Flux.never().subscribe(System.out::println);

//        Flux.range(10, 100).subscribe(System.out::println);

//        Flux.range(10, 100).subscribe(new Subscriber<Integer>() {
//            private Subscription subscription;
//
//            @Override
//            public void onSubscribe(Subscription s) {
//                subscription = s;
//                //要一个数据
//                subscription.request(1);
//            }
//
//            @Override
//            public void onNext(Integer integer) {
//                System.err.println(integer);
//                try {
//                    Thread.sleep(1000);
//                } catch (InterruptedException e) {
//                    throw new RuntimeException(e);
//                }
//                subscription.request(2);
//            }
//
//            @Override
//            public void onError(Throwable t) {
//                System.err.println("TestFlux中的方法onError 执行了--->");
//            }
//
//            @Override
//            public void onComplete() {
//                System.err.println("TestFlux中的方法onComplete 执行了--->");
//            }
//        });
//
//        Flux.create(t -> {
//            t.next("a").next("b").complete();
//        }).concatWithValues("c","d").subscribe(System.out::println);
//        Flux.just("1","2","3").concatWithValues("c","d").subscribe(System.out::println);
//       Flux.concat(Flux.just("1","2","3"),Flux.just("a","b","c")).subscribe(System.out::println);

//        Flux.concat(Flux.just(1, 2, 0)
//                        .doOnNext(integer -> System.out.println(100/integer))//doOnNext在执行next之前先执行的操作
//                , Flux.just(4, 5, 6)).subscribe(new Subscriber<Integer>() {
//            Subscription subscription;
//            @Override
//            public void onSubscribe(Subscription s) {
//                s.request(1);
//                subscription = s;
//            }
//
//            @Override
//            public void onNext(Integer integer) {
//                System.err.println(integer);
//                subscription.request(1);
//            }
//
//            @Override
//            public void onError(Throwable t) {
//                t.printStackTrace();
//            }
//
//            @Override
//            public void onComplete() {
//                System.err.println("TestMain中的方法onComplete 执行了");
//            }
//        });

//前面的数据失败的情况下可以继续执行
//        Flux.concatDelayError(Flux.just(1, 2, 0)
//                        .doOnNext(integer -> System.err.println(100/integer))//doOnNext在执行next之前先执行的操作
//                , Flux.just(4, 5, 6)).subscribe(new Subscriber<Integer>() {
//            Subscription subscription;
//            @Override
//            public void onSubscribe(Subscription s) {
//                s.request(1);
//                subscription = s;
//            }
//
//            @Override
//            public void onNext(Integer integer) {
//                System.err.println(integer);
//                subscription.request(1);
//            }
//
//            @Override
//            public void onError(Throwable t) {
//                t.printStackTrace();
//            }
//
//            @Override
//            public void onComplete() {
//                System.err.println("TestMain中的方法onComplete 执行了");
//            }
//        });
//        Flux.firstWithSignal(Flux.just(1, 2, 0) , Flux.just(4, 5, 6)).subscribe(new Subscriber<Integer>() {
//            Subscription subscription;
//            @Override
//            public void onSubscribe(Subscription s) {
//                s.request(1);
//                subscription = s;
//            }
//
//            @Override
//            public void onNext(Integer integer) {
//                System.err.println(integer);
//                subscription.request(1);
//            }
//
//            @Override
//            public void onError(Throwable t) {
//                t.printStackTrace();
//            }
//
//            @Override
//            public void onComplete() {
//                System.err.println("TestMain中的方法onComplete 执行了");
//            }
//        });
//        Flux.merge(Flux.interval(Duration.ofMillis(0), Duration.ofMillis(100)).take(5), Flux.interval(Duration.ofMillis(50), Duration.ofMillis(100)).take(5))
//                .toStream()
//                .forEach(System.out::println);
//结果就是两个flux的数据按照顺序输出自己的数据
//        Flux.mergeComparing((s1, s2)-> s1-s2, Flux.just(1, 3, 2), Flux.just(6, 5, 4)).subscribe(System.out::println);
//如果每个只有一个数据,可以看到明显的排序效果
//        Flux.mergeComparing((s1,s2)-> s2-s1, Flux.just(1), Flux.just(6), Flux.just(2), Flux.just(4), Flux.just(7)).subscribe(System.out::println);

//        Flux.zip(Flux.just(1, 2, 0,7), Flux.just(4, 5, 6,9)).subscribe(System.out::println);
//        Flux.just(1, 2, 0).all(i->i>-1).subscribe(System.out::println);
//        Flux.just(1, 2, 0).any(i->i>-0).subscribe(System.out::println);

//        System.err.println(""+Flux.just(1, 2, 3).as(integerFlux -> {
//            final String[] s = {""};
//            integerFlux.subscribe(integer -> s[0] = s[0] + integer);
//            return s[0];
//        }));

        //Flux.just(1,2,3,4,5).buffer(3).subscribe(System.out::println);
        //几个为1组,分完一组后跳过几个值再开始分组
//        Flux.just(1,2,3,4,5).buffer(2,1).subscribe(System.out::println);

//        Flux<Integer> cache = Flux.just(1, 2, 3, 4, 5).cache(2);
//        cache.subscribe(System.out::println);
////        //第二次次订阅后只有缓存的最后的个数的数据
//        cache.subscribe(System.out::println);

        //将最后指定个数的数据缓存,保存一定时间,在这个时间内订阅的话得到的是缓存的数据,当缓存时间结束后,再次订阅得到的结果就是是原始数据
//        Flux<Integer> cache1 = Flux.just(1, 2, 3, 4, 5).cache(2, Duration.of(10, ChronoUnit.SECONDS));
//        //第一次原始数据
//        cache1.subscribe(System.out::println);
//        System.out.println("-------");
//        //缓存的个数的数据
//        cache1.subscribe(System.out::println);
//        System.out.println("-------");
//        //缓存的个数的数据
//        cache1.subscribe(System.out::println);
//        System.out.println("-------");
//        //缓存的个数的数据
//        cache1.subscribe(System.out::println);
//        System.out.println("-------");
//        Thread.sleep(11000);
//        //过期后再订阅就是原始数据
//        cache1.subscribe(System.out::println);
//        Flux<Integer> stringFlux = Flux.just(1, 2, 3, 4, 5).cast(Integer.class);
//        stringFlux.subscribe(System.out::println);
//        Flux.just(1, 2, 3, 4, 5, 6, 6).collect(Collectors.toSet()).subscribe(System.out::println);
//        Flux.just(1, 2, 3, 4, 5, 6, 6).collect(
//                ()->"a",
//                (t,u)->System.out.println(u+t)
//        ).subscribe(System.err::println);
//
//        Mono<List<Integer>> collectList = Flux.just(1, 2, 3, 4, 5, 6, 6).collectList();
//        collectList.subscribe(System.err::println);
//        Flux.just().defaultIfEmpty(10).subscribe(System.out::println);
        //去重
      //  Flux.just(1, 2, 3, 4, 5,5).distinct().subscribe(System.out::println);

//        Flux.just(1, 2, 3, 4,5,5,5,5,5,5,5,5, 5,5,4,4,4,4,4,4,4,4,5).distinctUntilChanged().subscribe(System.out::println);
        //         //参数1,确定当前数据是不是已经存在过,符合这个条件的就是已经存在过了,比如我们例子中的条件是当前数据是不是>4
        //        //第一个数字1小于4,并且现在没有数据小于4,就保存起来,第二个数字2也小于4,但是因为已经有一个小于4的了,所以认为他俩是一样的
        //        //直到10之前都是不大于4,都是重复的数据,到了10之后是大于4的并且是第一个数据,所以保存起来,后面的数据有的大于有的小于等于,但是因为两个数据都已经有了,所以后面的数据全是重复的数据
        //        //参数2,用什么集合来保存这个重复的数据
//        Flux.just(1, 2, 3, 4, 4, 10, 5, 5, 4, 6, 7, 8, 9)
//                .distinct(integer -> integer >4, () -> new HashSet<>())
//                .subscribe(System.out::println);


//        Flux.just(1, 2, 3, 4, 5, 5)
//                .doOnNext(integer ->System.err.println( "准备给下一个数据了.要给的数据是:"+integer))
//                .doOnComplete(() -> System.err.println("马上结束"))//在onComplete执行之前执行
//                .subscribe(new Subscriber<Integer>() {
//                    Subscription subscription;
//
//                    @Override
//                    public void onSubscribe(Subscription subscription) {
//                        this.subscription = subscription;
//                        subscription.request(1);
//
//                    }
//
//                    @Override
//                    public void onNext(Integer integer) {
//                        System.err.println("TestFlux中的方法onNext 执行了" + integer);
//                        subscription.request(1);
//                    }
//
//                    @Override
//                    public void onError(Throwable throwable) {
//
//                    }
//
//                    @Override
//                    public void onComplete() {
//                        System.err.println("TestFlux中的方法onComplete 执行了");
//                    }
//                });

//        Flux.just(1, 2, 3, 4, 5, 5)
//                .filter(integer -> integer>3)
//                .subscribe(System.out::println);

//        Flux.just(1, 2, 3, 4, 5, 5)
//                .map(integer -> "a"+integer)
//                .subscribe(System.out::println);
//数据转换, 什么是数据转换?
//在我们阻塞的编程中 根据用户名查询到用户信息, 然后根据用户信息获取到它的订单数据,再根据订单数据计算出总价格
//对于响应式编程来说,只要出现了新的变量数据就是发生了转换
//比如一开始只有用户名,我们需要查询到用户数据,产生了一个新的变量用户数据,我们就认为现在数据从用户名变成了用户数据
//      Flux.just(1, 2, 3, 4, 5, 5)
//                .map(integer -> "a" + integer)
//              .map(s -> {
//                  ArrayList<String> list1 = new ArrayList<>();
//                  for (int i = 0; i < 5; i++) {
//                      list1.add(s + " jihe");
//                  }
//                  return list1;
//              }).flatMap(strings -> Flux.fromIterable(strings))//扁平化转换,最终会放到一起,我们可以用它来实现将一个集合中的每一个数据中的集合数据放到一起
//              .subscribe(System.err::println);

//
//        Flux.just(1, 2, 3, 4, 8, 5, 6, 6, 10, 100).doOnNext(integer -> {
//            System.err.println("当前拿到了" + integer);
//        }).doOnComplete(()->{
//            System.err.println("结束了");
//        }).then(Mono.just("abc")).subscribe(System.err::println);
//
        CountDownLatch latch = new CountDownLatch(1);
        latch.await();
    }
}
